package com.gaofei.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class MyConsumer {

    @Autowired
    private RedisTemplate redisTemplate;

    @KafkaListener(topics = "2102A",containerFactory="ackContainerFactory" )
    public void revMsg(ConsumerRecord<String, String> msgObject, Acknowledgment ack)  {
        String value = msgObject.value();
        String key = msgObject.key();
        Boolean b = redisTemplate.opsForValue().setIfAbsent(key, "",60000, TimeUnit.SECONDS);
        if(b){
            //Redission  --->看门狗
            //100s  --模拟业务的执行时长
            //存进去,获取到锁
            System.out.println("This is my message: "+value);

            //执行完之后把这个key给删除了
            redisTemplate.delete(key);
        }else {
            //随你怎么处理--->重复消费了.
        }
        //以上的业务逻辑执行完毕,告诉kafka 让它的offset(当前消费消息的指针)移动到下一位
        ack.acknowledge();
    }
}
